package com.qingcloud.df.executor.worker;
import com.qingcloud.base.remote.executor.ExecutorParam;
import com.qingcloud.df.sdk.component.config.ExecutorConfig;
import com.qingcloud.df.sdk.component.context.CompInstContext;
import com.qingcloud.df.sdk.component.context.ConfigManager;
import com.qingcloud.df.sdk.component.context.ConfigService;
import com.qingcloud.df.sdk.component.context.NodeManager;
import com.qingcloud.df.sdk.component.enums.EnvEnum;
import com.qingcloud.df.sdk.component.message.Ping;
import com.qingcloud.df.sdk.component.message.Pong;
import com.qingcloud.df.sdk.component.message.request.InstanceStartRequest;
import com.qingcloud.df.sdk.component.message.request.JobRequest;
import com.qingcloud.df.sdk.component.message.response.JobResponse;
import com.qingcloud.df.sdk.component.message.response.NodeContextReponse;
import com.qingcloud.df.sdk.component.transport.handler.ExecutorHandlerUtil;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Component
public class NodeService {

    @Resource
    private ConfigService configService;

    @Resource
    private ExecutorService executorService;

    public void init(){
        //获取配置文件
        requestJobConf();
        //启动所有应该启动的作业实例
        executorService.init();
        Map<Integer, List<CompInstContext>> jobMap  = getJobContextMap();
        NodeManager.refreshJobInstance(jobMap);
        NodeContextReponse response = new NodeContextReponse();
        response.setNodeId(ExecutorConfig.getAkkaNodeAddress());
        response.setContextMap(NodeManager.get());
        ExecutorHandlerUtil.sendNodeContext(response);
    }

    public void requestJobConf()  {
        //获取本地所有的Job
        Map<Integer, List<CompInstContext>> jobMap = getJobContextMap();
        JobRequest originJob = new JobRequest();
        originJob.setNodeId(ExecutorConfig.getAkkaNodeAddress());
        if(jobMap!=null && jobMap.size()>0){
            originJob.setJobList(jobMap.keySet().stream().collect(Collectors.toList()));
        }
        //从server端拉取job信息
        List<InstanceStartRequest> requests = ExecutorHandlerUtil.pull(originJob);
        if(requests!=null && requests.size()>0){
            List<ExecutorParam.InstanceParam> list= new ArrayList<>();
            List<Integer>  originJobIds = originJob.getJobList();
            for(InstanceStartRequest req: requests){
                //节点上已经存在的Job不用再次下载配置信息
                if(!originJobIds.contains(req.getJobId())) {
                    ExecutorParam.InstanceParam param = new ExecutorParam.InstanceParam();
                    param.setJobId(req.getJobId());
                    param.setInstanceId(req.getInstanceId());
                    param.setEnv(req.getEnv());
                    param.setConfigUrl(req.getConfigPath());
                    list.add(param);
                }
            }
            configService.refushConfig(list);
        }
        //整合集成，server和executor 维护的作业信息一致
        integration(requests);
    }

    private  Map<Integer, List<CompInstContext>>  getJobContextMap(){
        String env  = ExecutorConfig.getEnv();
        Map<Integer, CompInstContext> contextMap =  null;
        if(env.equalsIgnoreCase(EnvEnum.DEV.name)){
            contextMap =  ConfigManager.getDevConfig();
        } else{
            contextMap = ConfigManager.getProConfig();
        }
        Map<Integer, List<CompInstContext>> jobMap = ConfigManager.instanceConvertor(contextMap);
        return jobMap;
    }

    /**
     * 在当前的JobMap中，是否存在不存在的返回的JOB,如不存在需要在当前节点删除
     * @param requests
     */
    public  void integration(List<InstanceStartRequest> requests){
        List<Integer> newJobList = new ArrayList<>();
        String env  = ExecutorConfig.getEnv();
        Map<Integer, CompInstContext> instanceMap =  null;
        if(env.equalsIgnoreCase(EnvEnum.DEV.name)){
            instanceMap =  ConfigManager.getDevConfig();
        } else{
            instanceMap = ConfigManager.getProConfig();
        }
        if(requests==null || requests.size()==0){//从Server上为获取到节点信息，如果本地存在需要删除
            //无合适的Job需要启动 本地作业全部删除
            for(Map.Entry<Integer, CompInstContext> map : instanceMap.entrySet()) {
                CompInstContext context = map.getValue();
                executorService.destroyJob(context.getJobId(),context.getEnv());
            }
        }else{ //本地需要维护的作业
            for (InstanceStartRequest request: requests){
                newJobList.add(request.getJobId());
            }
            if(instanceMap!=null){
                for(Map.Entry<Integer, CompInstContext> map : instanceMap.entrySet()){
                    CompInstContext context = map.getValue();
                    if(!newJobList.contains(context.getJobId())){//Server端已经删除得作业，需要在工作节点上删除
                        executorService.destroyJob(context.getJobId(),context.getEnv());
                    }
                }
            }
        }
    }

    public static Pong serverPing(Ping ping){
        Pong pong = new Pong();
        pong.setKnownNode(true);
        pong.setNodeId(ExecutorConfig.getAkkaNodeAddress());//server 地址
        return pong;
    }

}
